package com.iussoft.receiveData.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apache.tools.DateUtils;
import com.apache.tools.MD5Utils;
import com.apache.tools.StrUtil;
import com.apache.tools.UUIDUtil;
import com.iussoft.manager.RedisManager;
import com.iussoft.model.DataParams;
import com.iussoft.receiveData.schedul.ScheduledDynamicTask;
import com.iussoft.receiveData.send.ReceiveDataManager;
import com.iussoft.receiveData.send.impl.DepositCertificateManager;
import com.iussoft.tools.JedisPoolFactory;
import com.iussoft.tools.LoggerUtil;
import com.iussoft.tools.SpringContextLoader;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

import java.util.Arrays;
import java.util.List;

/**
 * @description: 接入报文 kafka消费者接受消息topic=receiveMessage
 * @author: Hou Dayu
 * @date: Created in 2020/3/1
 */
public class KafakMessageServcie implements MessageListener<String,String> {

    private List<String> topic;
    private ScheduledDynamicTask dynamicTask;
    public KafakMessageServcie(String topic,ScheduledDynamicTask dynamicTask){
        this.topic = Arrays.asList(topic.split(","));
        this.dynamicTask = dynamicTask;
    }

    @Override
    public void onMessage(ConsumerRecord<String, String> data) {

        //根据不同主题，消费
        String msg = data.value();
        JSONObject json = JSON.parseObject(msg);
        String cacheKey = MD5Utils.MD5(json.getString("interfaceCode")+json.getString("sendTime"));
        if(StrUtil.isNull(String.valueOf(getRedisManager().getlObjCache(cacheKey, JedisPoolFactory.DATABASE)))){
            getRedisManager().createObjCache(cacheKey,"T",300, JedisPoolFactory.DATABASE);
        }else{
            return;
        }
        //  ReceiveDataManager manager = (ReceiveDataManager) SpringContextLoader.getBean("receiveDataManager");
        DepositCertificateManager manager = (DepositCertificateManager) SpringContextLoader.getBean("depositCertificateManager");
        DataParams params = new DataParams();
        params.setParams(json);
        String rowkey = manager.rollbackStr(cacheKey);
        try{
            params.setTraceId(rowkey);
            if(topic.contains(data.topic())){//接入报文:topic=receiveMessage
                if(StrUtil.isNull(json.getString("traceId"))){
                    params.setParams("traceId", UUIDUtil.getUUID("",32));
                }
                if(!"1".equals(json.getString("ifTest"))) {//档板数据不在存证
                    manager.receiveDataMsg(params);//处理信息存证
                   // manager.depositCertificate(params, 3);//处理信息存证
                }
                dynamicTask.submitRun(new Runnable() {//处理解析
                    ReceiveDataManager rmanager = (ReceiveDataManager) SpringContextLoader.getBean("receiveDataManager");
                    public void run() {
                        try {
                            rmanager.receiveDataMsg(params);
                        } catch (Exception e){
                            LoggerUtil.warnSource("exception = Exception; method = receiveDataMsg; traceId = {}; requestTime = {}; requestIp = {}; sourceMessage = {}; errorMessage = {}; exeSys = send; level = {}",params.getParamsToStr("traceId"), DateUtils.Now.fmt_yyyyMMdd_HHmmss(),params.getParamsToStr("requestUseIp"),msg.replaceAll("\n","").replaceAll("\r","").replaceAll(";","；"),e.getMessage(),"ERROR");
                        }
                    }
                });
            }
        } catch (Exception e){
            LoggerUtil.warnSource("exception = Exception; method = receiveMessage; traceId = {}; requestTime = {}; requestIp = {}; rowkey = {}; sourceMessage = {}; errorMessage = {}; exeSys = send; level = {}",params.getParamsToStr("traceId"), DateUtils.Now.fmt_yyyyMMdd_HHmmss(),params.getParamsToStr("requestUseIp"),rowkey,msg.replaceAll("\n","").replaceAll("\r","").replaceAll(";","；"),e.getMessage(),"ERROR");
        } finally {
            getRedisManager().delObjCache(cacheKey, JedisPoolFactory.DATABASE);
        }
    }

    private RedisManager getRedisManager(){
        RedisManager redisManager = (RedisManager) SpringContextLoader.getBean("redisManager");
        return redisManager;
    }
}
